package com.miao.im.support.mq;

import cn.hutool.core.util.IdUtil;
import cn.hutool.core.util.StrUtil;
import com.miao.im.config.RabbitBaseConfiguration;
import org.springframework.amqp.AmqpConnectException;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessageBuilder;
import org.springframework.amqp.core.MessageProperties;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.rabbit.core.RabbitTemplate.ConfirmCallback;
import org.springframework.amqp.rabbit.core.RabbitTemplate.ReturnsCallback;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.autoconfigure.condition.ConditionalOnClass;
import org.springframework.stereotype.Component;

/**
 * @Auther: miaoguoxin
 * @Date: 2021/3/11 15:56
 * @Description:
 */
@Component
@ConditionalOnClass(RabbitTemplate.class)
public class RabbitProducerMessageService implements IProducerMessageService {
    @Autowired
    private RabbitTemplate rabbitTemplate;

    @Autowired
    private RabbitBaseConfiguration rabbitBaseConfiguration;

    @Override
    public void send(String exchange, String routingKey, String json, ConfirmCallback confirmCallback, ReturnsCallback returnsCallback) {
        if (StrUtil.isEmpty(exchange) || null == json) {
            return;
        }
        String uuid = IdUtil.simpleUUID();
        Message message = MessageBuilder.withBody(json.getBytes())
                .setContentType(MessageProperties.CONTENT_TYPE_JSON)
                .setContentEncoding("UTF-8")
                .setMessageId(uuid)
                .build();
        if (confirmCallback == null) {
            confirmCallback = rabbitBaseConfiguration;
        }
        if (returnsCallback == null) {
            returnsCallback = rabbitBaseConfiguration;
        }
        this.rabbitTemplate.setReturnsCallback(returnsCallback);
        this.rabbitTemplate.setConfirmCallback(confirmCallback);

        //使用继承扩展的CorrelationData 、id消息流水号
        MqMessagePayload payload = new MqMessagePayload(uuid);
        payload.setMessage(message);
        payload.setExchange(exchange);
        payload.setRoutingKey(routingKey);
        //有可能长时间retry之后依然不能恢复Connection，如rabbitmq挂掉的情况，不能一直retry下去阻塞接口调用
        //这种情况是没有confirm的，因为消息都没有发出去。所以处理就更简单了
        //retry失败或者没有retry机制都会抛出AmqpConnectException，catch之后将消息保存起来即可
        try {
            rabbitTemplate.convertAndSend(exchange, routingKey, message, payload);
        } catch (AmqpConnectException e) {
            MqMessageFailedStore.add(payload);
        }
    }

    @Override
    public void send(String exchange, String routingKey, String json) {
        this.send(exchange, routingKey, json, null, null);
    }
}
